/**
 * File: FastSequenceBoxWriter.java
 * Created by: mhaimel
 * Created on: Jun 23, 2011
 * CVS:  $Id: FastSequenceBoxWriter.java 1.0 Jun 23, 2011 1:47:47 PM mhaimel Exp $
 */
package uk.ac.ebi.curtain.storage.sequence.fast;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.Closeable;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.sun.org.apache.bcel.internal.generic.NEW;

import uk.ac.ebi.curtain.storage.sequence.SequenceBox;
import uk.ac.ebi.curtain.utils.CurtainUncheckedException;
import uk.ac.ebi.curtain.utils.Pair;
import uk.ac.ebi.curtain.utils.TemplateBuilder;
import uk.ac.ebi.curtain.utils.concurrent.UtilsCollect;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;
import uk.ac.ebi.velvet.exec.RuntimeExec;

/**
 * @author mhaimel
 *
 */
public class FastSequenceBoxWriter  implements Closeable {
	public static final String SORT_CMD = "sort -k1,1n -S 1G $IN$ > $OUT$";
	private static final char SEPARATOR = '\t';
	private final Log log = LogFactory.getLog(this.getClass());
	private File file;
	private final AtomicBoolean sortedData = new AtomicBoolean();
	private final ReentrantLock lock = new ReentrantLock();
	private final AtomicReference<FastEntryWriter> out = new AtomicReference<FastEntryWriter>();
	private final AtomicReference<String> sortCmd = new AtomicReference<String>(SORT_CMD);

	private final int bufferSize;
	private final Queue<String> bufferQueue = UtilsCollect.newQueue();
	private final AtomicReference<FastSequenceBoxSortedWriter> writer = new AtomicReference<FastSequenceBoxSortedWriter>();
	public FastSequenceBoxWriter(File outputFile, boolean sortedData) {
		this(outputFile,sortedData,FastSequenceBoxSortedWriter.BUFFER_SIZE);
	}
	
	public FastSequenceBoxWriter(File file, boolean sortedData, int bufferSize) {
		this.file = file;
		this.bufferSize = bufferSize;
		setSortedData(sortedData);
	}
		
	private boolean isSortedData(){
		return this.sortedData.get();
	}
	
	private void setSortedData(boolean sorted){
		this.sortedData.set(sorted);
	}
	
	public void writeBuffered(Long id, SequenceBox entry) throws IOException{
		if(isSortedData()){
			throw new NotImplementedException("Not yet possible for sorted data - use " + FastSequenceBoxSortedWriter.class +" directly!");
		} else {
			doWrite(id,entry);
		}
	}
	
	public void write(Long id,SequenceBox entry) throws IOException{
		if(isSortedData()){
			this.getWriter().write(id, entry);
		} else {
			doWrite(id,entry);
		}
	}
	
	public FastSequenceBoxSortedWriter getWriter() {
		return writer.get();
	}
	
	private final AtomicBoolean isDrained = new AtomicBoolean(false);
	protected void doWrite(Long id, SequenceBox entry) throws IOException{
		if(!bufferQueue.add(_getOutputStream().getString(id, entry))){
			throw new CurtainUncheckedException("Adding value to queue unsuccessfull!!!" + bufferQueue.size());
		}
		if(bufferQueue.size() >= 10000){ // ~3MB buffer
			// >= required, in case it fills faster than it gets drained
			if(isDrained.compareAndSet(false, true)){ // ensure only one drains the buffer at a time
				_drainBuffer(10000);
				isDrained.set(false);
			}
		}
	}
	
	
	private void _drainBuffer(int cnt) throws IOException {
		lock.lock();
		try{
			PrintWriter out = _getOutputStream().getWriter();
			String tmp = null;
			while(cnt>0){ // exact cnt not needed
				tmp = bufferQueue.poll();
				if(null == tmp){
					// done - empty queue
					break;
				}
				out.write(tmp);
				--cnt;
			}
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}

	public static char getSeparator(){
		return SEPARATOR;
	}
	
	private FastEntryWriter _getOutputStream() throws IOException{
		return out.get();
	}
	
	public void open()throws IOException {
		lock.lock();
		try{
			if(isSortedData()){
				this.writer.set(_buildSortedWriter());
				this.getWriter().open();
			} else {
				FastEntryWriter o = out.get();
				if(o != null){
					throw new IOException("File already in usage!!!"  + getUnsortedFile());
				}
				o = new FastEntryWriter(_doBuildStream(),getSeparator());
				out.set(o);
			}
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}

	protected FastSequenceBoxSortedWriter _buildSortedWriter() {
		return new FastSequenceBoxSortedWriter(file,bufferSize);
	}

	protected void closeOutputStream() throws IOException{
		lock.lock();
		try{
			if(isSortedData()){
				this.getWriter().close();
			} else {
				FastEntryWriter o = out.getAndSet(null);
				if(null == o ){
					throw new IOException("File not opened: "  + getUnsortedFile());
				}
				o.getWriter().close();
			}
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}
		
	protected PrintWriter _doBuildStream() throws IOException {
		File uFile = getUnsortedFile();
		if(uFile.exists()){
			throw new IOException("File already exists!!!"  + uFile);
		}
		return new PrintWriter(new BufferedWriter(new FileWriter(uFile,true),getBufferSize()));
	}
	
	public File getFile() {
		return file;
	}
	protected File getUnsortedFile(){
		return new File(this.getFile().getAbsolutePath()+".unsorted.txt");
	}
	
	protected int getBufferSize() {
		return this.bufferSize ;
	}

	public Log getLog() {
		return log;
	}

	@Override
	public void close() throws IOException {
		if(!isSortedData()){
			getLog().debug("Drain buffer ... ");
			this._drainBuffer(this.bufferQueue.size());
		}
		getLog().debug("Close writer ...");
		this.closeOutputStream();
	}
	
	public void sort()  throws IOException {
		getLog().debug("Sort data of file " + this.getUnsortedFile());
		File uFile = getUnsortedFile();
		_preSortCheck(uFile);
		setSortedData(true);
		_executeSort(uFile);
		try{
			getLog().debug("Sort of data done - start loading to" + this.getFile());
			this.open();			
			reprocess(uFile);
			this.getWriter().close();
		}finally{
			FileIO.closeQuietly(this);
		}
		getLog().debug("Sorting and loading done: " + this.getFile());
	}

	protected void _preSortCheck(File uFile) throws IOException, FileNotFoundException {
		if(isSortedData()){
			throw new IllegalStateException("Data already sorted!!!");
		}
		if(null != _getOutputStream()){
			throw new IllegalStateException("Close Writer first before sorting!!!");
		}
		if(!uFile.exists()){
			throw new FileNotFoundException("Unsorted data file not found: " + uFile);
		}
		if(uFile.length() == 0){
			throw new FileNotFoundException("Unsorted data file is empty: " + uFile);
		}
	}

	private void reprocess(File uFile) throws IOException {
		BufferedReader in = null;
		try {
			in = _openFile(uFile);
			String line = null;
			while((line = in.readLine())!=null){
				Pair<Long,SequenceBox> entry = FastSequenceBoxReader._parseEntry(line);
				this.getWriter().write(entry.getA(), entry.getB());
			}
		} finally{
			FileIO.closeQuietly(in);
		}
		FileUtils.deleteQuietly(uFile);
	}

	protected BufferedReader _openFile(File uFile) throws FileNotFoundException {
		return new BufferedReader(new FileReader(uFile));
	}

	protected void _executeSort(File uFile) throws IOException {
		File dir = uFile.getParentFile();
		File tmp = new File(dir,"tmp-"+uFile.getName());
		File orig = new File(dir,"orig-"+uFile.getName());

		RuntimeExec exec = new RuntimeExec(dir);
		StringBuffer out = new StringBuffer();
		StringBuffer err = new StringBuffer();
		String cmd = buildSortCommand(uFile,tmp);
		int retVal = exec.exec(cmd,out,err);
		if(retVal != 0){
			throw new IOException("Exit with " + retVal+" of command:" + cmd+"! Err: " + err.toString());
		}
		FileUtils.moveFile(uFile, orig);
		FileUtils.moveFile(tmp, uFile);
		orig.delete();
	}
	
	public String getSortCmd() {
		return sortCmd.get();
	}
	public void setSortCmd(String cmd) {
		sortCmd.set(cmd);
	}
	private String buildSortCommand(File input,File tmpFile) {
		String cmd = getSortCmd();
		TemplateBuilder tmp = new TemplateBuilder(cmd);
		tmp.addPlaceHolder("IN", input.getAbsolutePath());
		tmp.addPlaceHolder("OUT", tmpFile.getAbsolutePath());
		return tmp.generate();
	}
}
